package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.util.DimUtil;
import com.atguigu.gmall.realtime.util.HbaseUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;

/**
 * @author Felix
 * @date 2023/8/21
 * 将流中的数据写到Hbase对应的表中
 */
public class DimSinkFunction extends RichSinkFunction<JSONObject> {

    private Connection conn;

    @Override
    public void open(Configuration parameters) throws Exception {
        conn = HbaseUtil.getHbaseConnection();
    }

    @Override
    public void close() throws Exception {
        HbaseUtil.closeHbaseConnection(conn);
    }

    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        // {"tm_name":"苹果","id":2,"type":"update","table_process":{"op":"r","sinkColumns":"id,tm_name",
        // "sinkFamily":"info","sinkRowKey":"id","sinkTable":"dim_base_trademark","sourceTable":"base_trademark"}}
        //获取对业务数据库维度表的操作的类型
        String type = jsonObj.getString("type");
        //获取当前操作的这条维度数据对应的配置表信息
        TableProcess tableProcess = jsonObj.getObject("table_process", TableProcess.class);
        //获取操作的hbase中表名
        String sinkTable = tableProcess.getSinkTable();
        //获取rowkey
        String rowKeyName = tableProcess.getSinkRowKey();
        String rowKey = jsonObj.getString(rowKeyName);

        //判断操作类型
        if ("delete".equals(type)) {
            //从hbase的维度表中删除一条数据
            System.out.println("从hbase的" + sinkTable + "表中删除" + rowKey + "数据");
            HbaseUtil.delRow(conn, GmallConfig.HBASE_NAMESPACE, sinkTable, rowKey);
        } else {
            System.out.println("向hbase的" + sinkTable + "表中put" + rowKey + "数据");
            //insert|update 向hbase的维度表中put一条数据
            String family = tableProcess.getSinkFamily();
            String[] columnNames = tableProcess.getSinkColumns().split(",");
            String[] columnValues = new String[columnNames.length];
            for (int i = 0; i < columnNames.length; i++) {
                String columnName = columnNames[i];
                columnValues[i] = jsonObj.getString(columnName);
            }

            HbaseUtil.putRow(conn, GmallConfig.HBASE_NAMESPACE, sinkTable, rowKey, family, columnNames, columnValues);
        }

        //如果维度数据发生了变化，将Redis中缓存的维度数据清除掉
        if("delete".equals(type)||"update".equals(type)){
            DimUtil.delCached(sinkTable,rowKey);
        }

    }
}
